package com.zhichat.infrastructure.thread;

import lombok.Getter;
import lombok.extern.slf4j.Slf4j;

import java.util.Optional;
import java.util.concurrent.*;
import java.util.concurrent.ThreadPoolExecutor.CallerRunsPolicy;

@Getter
@Slf4j
public class ThreadPoolWorker {

    private final ThreadPoolExecutor executor;
    public static final int DEFAULT_POOL_SIZE = Optional
            .ofNullable(System.getProperty("reactor.schedulers.defaultPoolSize"))
            .map(Integer::parseInt).orElseGet(() -> Runtime.getRuntime().availableProcessors());

    private ThreadPoolWorker() {
        this.executor = new ThreadPoolExecutor(DEFAULT_POOL_SIZE * 2,
                DEFAULT_POOL_SIZE * 3,
                5,
                TimeUnit.MINUTES,
                new ArrayBlockingQueue<Runnable>(DEFAULT_POOL_SIZE * 2),
                new CallerRunsPolicy());
        log.debug("初始化线程池，线程池核心大小:{},最大线程数:{},等待队列大小:{}", executor.getCorePoolSize(), executor.getMaximumPoolSize(), DEFAULT_POOL_SIZE * 2);
    }

    public void shutdown() {
        log.debug("消费自定义线程池.....");
    }

    public static ThreadPoolWorker singleInstance() {
        return Single.single;
    }

    static class Single {
        private static final ThreadPoolWorker single = new ThreadPoolWorker();
    }

    class CustomerRejectedExecutionHandler implements RejectedExecutionHandler {
        private final long maxWait = 2000;

        @Override
        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
            if (!executor.isShutdown()) {
                try {
                    BlockingQueue<Runnable> queue = executor.getQueue();
                    if (log.isDebugEnabled()) {
                        log.debug("Attempting to queue task execution for " + this.maxWait
                                + " milliseconds");
                    }
                    if (!queue.offer(r, this.maxWait, TimeUnit.MILLISECONDS)) {
                        throw new RejectedExecutionException("Max wait time expired to queue task");
                    }
                    if (log.isDebugEnabled()) {
                        log.debug("Task execution queued");
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    throw new RejectedExecutionException("Interrupted", e);
                }
            } else {
                throw new RejectedExecutionException("Executor has been shut down");
            }
        }
    }
}
